package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author jiwei
 * @description 模拟热力数据写入Kafka
 * @date 2023/6/12 9:02
 */
public class Datagen2Kafka_Heating {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //设置时区
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));

        //Source
        tableEnv.executeSql("CREATE TABLE source_people_info (\n" +
                "     id INT,\n" +
                "     hobby STRING,\n" +
                "     school STRING,\n" +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (\n" +
                "     'connector' = 'mysql-cdc',\n" +
                "     'hostname' = '192.168.102.155',\n" +
                "     'port' = '33066',\n" +
                "     'username' = 'datacube_test',\n" +
                "     'password' = 'Gbd#XERA@19d1ZtD',\n" +
                "     'database-name' = 'db_datacube_test',\n" +
                "     'table-name' = 'sorce_people_info')");

        //Sink
        tableEnv.executeSql("CREATE TABLE sink_people_info (\n" +
                "     id INT,\n" +
                "     hobby STRING,\n" +
                "     school STRING,\n" +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (\n" +
                "     'connector' = 'upsert-kafka',\n" +
                "     'topic' = 'yzq_test',\n" +
                "     'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "     'key.format' = 'json',\n" +
                "     'value.format' = 'json')");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sink_people_info(id, hobby, school)\n" +
                "SELECT \n" +
                "id,\n" +
                "hobby,\n" +
                "school\n" +
                "FROM source_people_info");

    }
}
